package com.spark.zhou.demo.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * @Description: 官网demo测试
 * @Author: ZhOu
 * @Date: 2018/11/6
 */
public class StreamingSocketDemo {
    public static void main(String[] args) throws InterruptedException {
        //此处的master线程必须大于1。如果只有一个线程，则单线程将用于运行接收器，就没有线程来处理接收到的数据
        SparkConf sparkConf = new SparkConf()
                .setAppName("streamingDemo")
                .setMaster("local[2]");

        //批处理时间间隔为10秒
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(10));

        //创建一个DStream,表示来自TCP源的流数据，指定地址和端口
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("192.168.66.11", 9999);

        //从数据服务器接收数据流，流中每条纪录都是文本，按空格分割单词
        JavaDStream<String> words = lines.flatMap(v -> Arrays.asList(v.split(" ")).iterator());

        //转为pair，以单词为键，以1为值
        JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));

        //以键名统计单词出现次数
        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((m, n) -> m + n);

        System.out.println("\n\n=================================");
        wordCounts.print();
        System.out.println("=================================\n\n");
        //开始处理
        jssc.start();

        jssc.awaitTermination();
    }
}
